package com.qdone.framework.util.task;


import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.linkedin.parseq.EngineBuilder;
import com.linkedin.parseq.batching.BatchingSupport;

import java.util.concurrent.*;


/**
 * Task的执行者工厂类
 */
public class EngineFactory {

    // 单例模式
    private static EngineFactory INSTANCE = new EngineFactory();
    private final EngineAgent defaultEngine;

    // 初始化任务执行者
    private EngineFactory() {
        // Java虚拟机的可用的处理器数量
        int numCores = Runtime.getRuntime().availableProcessors();
        // 默认 任务执行线程池大小 与 任务调度池程池大小
        defaultEngine = getEngine(numCores + 1, 1, numCores + 1);
    }

    public static EngineAgent defaultEngine() {
        return INSTANCE.defaultEngine;
    }

    /**
     * 初始化异步处理框架
     *
     * @param poolSize     任务执行线程池大小
     * @param scheduleSize 任务调度池程池大小
     * @param queueNum     任务对列数量
     * @return
     */
    public static EngineAgent getEngine(int poolSize, int scheduleSize, int queueNum) {
        ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(scheduleSize);
        ThreadFactory threadFactory = new ThreadFactoryBuilder().build();
        ThreadPoolExecutor executors = new ThreadPoolExecutor(poolSize, poolSize,
                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(queueNum), threadFactory,
                (r, executor) -> {
                    try {
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });

        final EngineBuilder builder = new EngineBuilder().setTaskExecutor(executors).setTimerScheduler(scheduler);
        final BatchingSupport batchingSupport = new BatchingSupport();
        // 批处理
        builder.setPlanDeactivationListener(batchingSupport);
        return new EngineAgent(builder.build(), executors, scheduler);
    }

}
